package com.tang.window;

import com.tang.bean.WaterSensor;
import com.tang.functions.WaterSensorMapFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * WindowApi
 *
 * @author tang
 * @since 2023/6/11 9:04
 */
public class WindowApiDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> source = env
                .socketTextStream("192.168.70.141", 7777)
                .map(new WaterSensorMapFunction());
        
        KeyedStream<WaterSensor, String> keyedStream = source.keyBy(WaterSensor::getId);

        /**
         * 1. 指定窗口分配器
         * 指定用哪一种窗口？ --- 时间 or 计数？滚动、滑动、会话？
         */
        // 1.1 没有keyBy的窗口 窗口内的 所有数据 进入同一个 子任务，并行度只能为1
        // source.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)));
        // 1.2 keyBy的窗口 每个key上都定义了一组窗口，各自独立地进行统计计算
        // 1.2.1 基于时间的
        // (1)滚动处理时间窗口 ，每10s一个窗口
//        keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
        // (2)滑动时间处理窗口，每5秒启动一个窗口，窗口时间是10s
//        keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)));
        // (3)会话窗口，超时间隔11s
//        keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(11)));

        // 1.2.2 基于计数的
        // (1) 滚动窗口，窗口长度=10个元素
//        keyedStream.countWindow(10);
        // (2) 滑动窗口，窗口长度=5个元素，滑动步长=2个元素
//        keyedStream.countWindow(5, 2);
        // (3) 全局窗口，计数窗口的底层就是用的这个，需要自定义的时候才会用
//        keyedStream.window(GlobalWindows.create());
        
        // 2. 指定窗口函数：窗口内数据的计算逻辑
        WindowedStream<WaterSensor, String, TimeWindow> window = keyedStream
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
        // 增量聚合 来一条计算一条，窗口触发的时候输出计算结果
//        sensorWS
//                .reduce()
//        .aggregate(, )
        
        // 全窗口函数 数据来了不计算，存起来，窗口触发的时候，计算并输出结果
//        sensorWS.process()
        
        env.execute();

    }

}
